package cn.doitedu.demo6;

import cn.doitedu.demo1.UserEvent_1;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/20
 * @Desc: 学大数据，上多易教育
 *    运算机模型类：
 *    支持的规则如下：
 *    // 年龄>25岁,且 性别 = male 的用户 ，规则生效后，做过 C行为3次以上，当他发生 K 行为，则命中规则
 *
 *    // 年龄<35岁,且 性别 = female ,且 活跃等级>2    的用户 ，规则生效后，做过 X行为5次以上，当他发生 P 行为，则命中规则
 *
 *    // 年龄<35岁,且 性别 = female ,且 活跃等级>2    的用户 ，规则生效后，做过 S行为2次以上，当他发生 P 行为，则命中规则
 *
 *
 *
 **/
public class RuleModel_2_Calculator implements RuleCalculator{

    Model2Param ruleParam;
    ValueState<Integer> cntState;
    Table htable;
    JSONObject message = new JSONObject();

    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJson) throws IOException {

        // 解析规则参数
        ruleParam = JSON.parseObject(ruleParamJson, Model2Param.class);

        // 申请flink状态
        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleParam.getRule_id(), Integer.class));

        // 创建hbase客户端连接
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","doitedu:2181");
        Connection connection = ConnectionFactory.createConnection(conf);
        htable = connection.getTable(TableName.valueOf("user_profile"));

    }

    @Override
    public void calc(UserEvent_1 eventBean, Collector<String> out) throws IOException {

        // 动态画像统计
        if(eventBean.getEvent_id().equals(ruleParam.getRealtime_profile().getEvent_id())){
            cntState.update(cntState.value()==null?1: cntState.value()+1);
        }


        // 触发判断
        if(eventBean.getEvent_id().equals(ruleParam.getFire_condition().getEvent_id())){

            // 1. 先判断动态画像条件是否已经满足
            if(cntState.value()!=null && cntState.value()> ruleParam.getRealtime_profile().getMin_count()  ){

                // 2. 再判断静态画像标签是否满足条件

                // 根据规则参数，拼接hbase查询条件
                Get get = new Get(Bytes.toBytes(eventBean.getUid()));
                List<ProfileCondition> profileConditions = ruleParam.getProfile_conditions();
                for (ProfileCondition profileCondition : profileConditions) {
                    get.addColumn( Bytes.toBytes("f"),   Bytes.toBytes(profileCondition.getTag_name()));
                }

                // 执行hbase查询
                Result result = htable.get(get);

                // 遍历每一个画像条件，进行满足与否的判断
                for (ProfileCondition profileCondition : profileConditions) {
                    // 取出规则条件中的一个条件标签，对应在hbase结果中的值
                    byte[] queryValueBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes(profileCondition.getTag_name()));
                    // 判断查询出来的值，是否满足规则条件中要求的值
                    String compareType = profileCondition.getCompare_type();
                    List<String> conditionValues = profileCondition.getCondition_values();

                    boolean res = compare(Bytes.toString(queryValueBytes), conditionValues, compareType);

                    if(!res) return;
                }

                // 如果能到这里，说明每一个静态画像条件都满足了
                // 该推送命中消息了
                message.put("rule_id",ruleParam.getRule_id());
                message.put("user_id",eventBean.getUid());
                message.put("hit_time",eventBean.getTimestamp());

                out.collect(message.toJSONString());


            }

        }

    }




    private boolean compare(String queryValue,List<String> conditionValues,String compareType){

        boolean res = false;

        switch (compareType) {
            case ">":
               res = Double.parseDouble(queryValue) > Double.parseDouble(conditionValues.get(0));
                break;
            case ">=":
                res = Double.parseDouble(queryValue) >= Double.parseDouble(conditionValues.get(0));
                break;
            case "<":
                res = Double.parseDouble(queryValue) < Double.parseDouble(conditionValues.get(0));
                break;
            case "<=":
                res = Double.parseDouble(queryValue) <= Double.parseDouble(conditionValues.get(0));
                break;
            case "=":
                res = queryValue.equals(conditionValues.get(0));
                break;
            case "between":
                double queryV = Double.parseDouble(queryValue);
                double conditionV1 = Double.parseDouble(conditionValues.get(0));
                double conditionV2 = Double.parseDouble(conditionValues.get(1));
                res = queryV>=conditionV1 && queryV<=conditionV2;
                break;
            default:
                res = false;

        }

        return res;
    }



}
